package com.swak.job.worker;

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionStage;
import java.util.function.Function;

import com.swak.job.Job;
import com.swak.job.JobConfig;
import com.swak.job.JobTask;
import com.swak.job.enums.Role;
import com.swak.job.enums.Status;
import com.swak.job.queue.QueueService;
import com.swak.meters.MetricsFactory;
import com.swak.registry.RegistryService;
import com.swak.utils.JsonMapper;
import com.swak.utils.time.DateTimes;

/**
 * 任务执行
 * 
 * @author DELL
 */
public class ExecuteWorker extends AbstractWorker {

	/**
	 * 处理消费请求
	 */
	Function<JobTask, Object> jobTaskConsumeFunc = this::handleJobTaskRequest;

	public ExecuteWorker(JobConfig config, Job job, QueueService queueService, RegistryService registryService,
			MetricsFactory metricsFactory) {
		super(config, Role.Executer, job, queueService, registryService, metricsFactory);
	}

	@Override
	public void start() {
		super.start();
		this.queueService.createQueue(queue);
		int parallel = this.getRefreshAbleParallel();
		for (int i = 0; i < parallel; i++) {
			this.queueService.consumeQueue(queue, jobTaskConsumeFunc);
		}
	}

	private Object handleJobTaskRequest(JobTask task) {
		Object metrics = this.namedMetrics != null ? this.namedMetrics.begin() : null;
		try {
			if (logger.isDebugEnabled()) {
				logger.debug("Job {} Execute Task {} Start.", this.job.getName(), task.getConfig());
			}
			Object[] args = new Object[] { task };
			Object result = this.invoke(args);
			return this.executeTask(task, result, metrics);
		} catch (Throwable e) {
			CompletableFuture<Void> future = new CompletableFuture<>();
			this.executeResult(future, task, null, e, metrics);
			return future;
		}
	}

	/**
	 * 执行任务
	 * 
	 * @param result
	 */
	@SuppressWarnings("unchecked")
	protected CompletableFuture<Void> executeTask(JobTask task, Object result, Object metrics) {
		CompletableFuture<Void> future = new CompletableFuture<>();
		if (result != null && result instanceof CompletionStage) {
			CompletionStage<Object> resultFuture = (CompletionStage<Object>) result;
			resultFuture.whenComplete((r, e) -> {
				this.executeResult(future, task, r, e, metrics);
			});
		} else {
			this.executeResult(future, task, result, null, metrics);
		}
		return future;
	}

	protected void executeResult(CompletableFuture<Void> future, JobTask task, Object result, Throwable e,
			Object metrics) {
		boolean success = true;
		try {
			if (logger.isDebugEnabled()) {
				logger.debug("Job {} Execute Task {} End, Execute {}.", this.job.getName(), task.getConfig(),
						e == null ? Status.Success : Status.Failure);
			}
			if (e != null) {
				logger.error("Job {} Execute Task {} Error.", this.job.getName(), task.getConfig(), e);
			}
			task.setResult(JsonMapper.toJson(result));
			task.setStatus(e == null ? Status.Success.getValue() : Status.Failure.getValue());
			task.setLastExecutionTime(DateTimes.now());
			// this.storageService.execute(task);
			
			// 向调度器响应执行结果
			// 消费不成功或者响应不成功 - 都会发起重试
			String query = this.queueService.parseDispatchQueue(this.url);
			this.queueService.publishQueueAsync(query, task).whenComplete((r, ex) -> {
				if (e != null || ex != null) {
					future.completeExceptionally(e);
				} else {
					future.complete(null);
				}
			});
		} catch (Exception e2) {
			success = false;
			future.completeExceptionally(e2);
		} finally {
			if (this.namedMetrics != null) {
				this.namedMetrics.end(metrics, success);
			}
		}
	}

	@Override
	public void stop() {
		super.stop();
		this.queueService.unconsumeQueue(queue, jobTaskConsumeFunc);
	}
}
